# -*- coding: utf-8 -*-
"""
airflow api supplement
"""
import logging

# import airflow.api
from airflow import settings
from airflow.models import DagRun

from airflow.plugins_manager import AirflowPlugin
from airflow.utils.helpers import alchemy_to_dict

from flask import Blueprint, jsonify

_log = logging.getLogger(__name__)

# requires_authentication = airflow.api.api_auth.requires_authentication

rest_api_bp = Blueprint('api-plugin', __name__, url_prefix='/api/plugin')


# define plugin
class ApiPlugin(AirflowPlugin):
    """
    api plugin
    """
    name = "rest_api"
    operators = []
    flask_blueprints = [rest_api_bp]
    hooks = []
    executors = []
    admin_views = []
    menu_links = []


# api
@rest_api_bp.route('/dags/<string:dag_id>/runs/<string:run_id>', methods=['GET'])
# @requires_authentication
def dag_run_info(dag_id, run_id):
    """Returns a JSON with a dag run's info. """
    from airflow.www.views import dagbag

    if dag_id not in dagbag.dags:
        response = jsonify(error='Dag {} not found'.format(dag_id))
        response.status_code = 404
        return response

    session = settings.Session()
    dag_runs = DagRun.find(dag_id, run_id)
    if not dag_runs:
        response = jsonify(error='Dag {} Run {} not found'.format(dag_id, run_id))
        response.status_code = 404
        return response

    dag_run = dag_runs[0]

    dag = dagbag.get_dag(dag_id)
    dttm = dag_run.execution_date
    tis = dag.get_task_instances(session, dttm, dttm)

    dr = alchemy_to_dict(dag_run)
    dr['task_instances'] = {
        ti.task_id: alchemy_to_dict(ti)
        for ti in tis
    }

    return jsonify(dr)
